package com.google.code.p.restrpc.directory;

import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.code.p.restrpc.DefaultServiceNode;
import com.google.code.p.restrpc.ServiceNode;
import com.google.code.p.restrpc.directory.heartbeat.RegistryHeartBeater;
import com.google.code.p.restrpc.event.RpcEventCenter;
import com.google.code.p.restrpc.event.RpcEventListener;
import com.google.code.p.restrpc.event.RpcEventObject;

/**
 * @author dz
 */
public class DefaultRegistry implements Registry {
    
    private static final Log logger = LogFactory.getLog(DefaultRegistry.class);
    
    private ConcurrentHashMap<String, DefaultServiceNode> serverNodes = new ConcurrentHashMap<String, DefaultServiceNode>();
    
    private ConcurrentHashMap<String, ServiceDefinitionGroup> services = new ConcurrentHashMap<String, ServiceDefinitionGroup>();
    
    private RegistryHeartBeater heatBeatMonitor;
    
    /**
     * 
     */
    public DefaultRegistry() {
        this(60 * 1000);
    }
    
    /**
     * 
     */
    public DefaultRegistry(int heartbeat) {
        register(new DefaultRegistryListener(this));
        heatBeatMonitor = new RegistryHeartBeater(this, heartbeat);
        heatBeatMonitor.start();
    }
    
    /**
     * 
     */
    public void disable(RemoteServiceDefinition rsd) {
        String serviceName = rsd.getServiceName();
        if (StringUtils.isEmpty(serviceName)) {
            throw new IllegalArgumentException("The 'serviceName' can't be null or empty!");
        }
        
        ServiceDefinitionGroup sdg = services.get(rsd.getServiceName());
        if (sdg == null) {
            return;
        }
        
        sdg.enable(rsd);
    }

    /**
     * 
     */
    public void enable(RemoteServiceDefinition rsd) {
        String serviceName = rsd.getServiceName();
        if (StringUtils.isEmpty(serviceName)) {
            throw new IllegalArgumentException("The 'serviceName' can't be null or empty!");
        }
        
        ServiceDefinitionGroup sdg = services.get(rsd.getServiceName());
        if (sdg == null) {
            return ;
        }
        sdg.disable(rsd);
    }
    /**
     * @return
     */
    public String[] getNodeAddresses() {
        return (String[]) serverNodes.keySet().toArray(new String[] {});
    }
    
    /**
     * @return
     */
    public String[] getServiceNames() {
        return (String[]) services.keySet().toArray(new String[] {});
    }
    
    
    private NodeEventObject changeNodeStatus(RemoteServiceDefinition rsd,
                                             boolean status) {
        String serviceNodeKey = rsd.getRemoteAddress();
        if (StringUtils.isEmpty(serviceNodeKey)) {
            throw new IllegalArgumentException("The remoteAddress can't be null or empty!");
        }
        if (rsd.getRemotePort() != 80) {
            serviceNodeKey = new StringBuilder(rsd.getRemoteAddress()).append(':')
                                                                     .append(rsd.getRemotePort())
                                                                     .toString();
        }
        ServiceNode sn = serverNodes.get(serviceNodeKey);
        if (sn == null) {
            return null;
        }
        
        boolean previous = status ? sn.enable() : sn.disable();
        if (previous == status) {
            return null;
        }
        
        return new NodeEventObject(sn, previous, status);
    }
    
    /**
     * 
     */
    public void register(RemoteServiceDefinition rsd) {
        String serviceName = rsd.getServiceName();
        if (StringUtils.isEmpty(serviceName)) {
            throw new IllegalArgumentException("The 'serviceName' can't be null or empty!");
        }
        
        ServiceDefinitionGroup sdg = services.get(rsd.getServiceName());
        if (sdg == null) {
            services.putIfAbsent(serviceName,
                                 new ServiceDefinitionGroup(serviceName));
            sdg = services.get(serviceName);
        }
        
        String serviceNodeKey = rsd.getRemoteAddress();
        if (StringUtils.isEmpty(serviceNodeKey)) {
            throw new IllegalArgumentException("The 'remoteAddress' can't be null or empty!");
        }
        if (rsd.getRemotePort() != 80) {
            serviceNodeKey = new StringBuilder(rsd.getRemoteAddress()).append(':')
                                                                     .append(rsd.getRemotePort())
                                                                     .toString();
        }
        
        DefaultServiceNode sn = serverNodes.get(serviceNodeKey);
        if (sn == null) {
            serverNodes.putIfAbsent(serviceNodeKey, new DefaultServiceNode());
            sn = serverNodes.get(serviceNodeKey);
            sn.setAddress(rsd.getRemoteAddress());
            sn.setPort(rsd.getRemotePort());
            sn.setWeight(rsd.getWeight());
        }
        else {
            if (!sn.isAvailable()) {
                enable(rsd);
            }
        }
        sn.setLastTimeMillis(System.currentTimeMillis());
        
        sdg.add(rsd, sn);
        
        if (logger.isDebugEnabled()) {
            logger.debug("Remote Service Registered:" + rsd.toString());
        }
    }
    
    /**
     * 
     */
    public void unregister(RemoteServiceDefinition rsd) {
        String serviceName = rsd.getServiceName();
        if (StringUtils.isEmpty(serviceName)) {
            throw new IllegalArgumentException("The serviceName can't be null or empty!");
        }
        
        ServiceDefinitionGroup sdg = services.get(rsd.getServiceName());
        if (sdg == null) {
            return;
        }
        
        sdg.remove(rsd);
        

        if (logger.isDebugEnabled()) {
            logger.debug("Remote Service Unregistered:" + rsd.toString());
        }
    }
    
    /**
     * @param address
     * @param port
     */
    public void activate(String address, int port) {
        if (StringUtils.isEmpty(address)) {
            throw new IllegalArgumentException("The 'address' can't be null or empty!");
        }
        
        String serviceNodeKey = address;
        if (port != 80) {
            serviceNodeKey = new StringBuilder(address).append(':')
                                                      .append(port)
                                                      .toString();
        }
        
        DefaultServiceNode serviceNode = serverNodes.get(serviceNodeKey);
        if (serviceNode == null) {
            if (logger.isWarnEnabled()) {
                logger.warn(serviceNodeKey + " not found!");
            }
            return;
        }
        serviceNode.enable();
        //TODO if serviceNode is enabled, the client need to upgrade the lookup result
        serviceNode.setLastTimeMillis(System.currentTimeMillis());
    }
    
    /**
     * 
     */
    public boolean isRegistered(String address, int port) {
        if (StringUtils.isEmpty(address)) {
            throw new IllegalArgumentException("The 'address' can't be null or empty!");
        }
        
        String serviceNodeKey = address;
        if (port != 80) {
            serviceNodeKey = new StringBuilder(address).append(':')
                                                      .append(port)
                                                      .toString();
        }
        
        return serverNodes.containsKey(serviceNodeKey);
    }
    
    /**
     * 
     */
    public void unregisterAll(String address, int port) {
        if (StringUtils.isEmpty(address)) {
            throw new IllegalArgumentException("The 'address' can't be null or empty!");
        }
        
        //
        String serviceNodeKey = address;
        if (port != 80) {
            serviceNodeKey = new StringBuilder(address).append(':')
                                                      .append(port)
                                                      .toString();
        }
        if (logger.isDebugEnabled()) {
            logger.debug("Unregistered Node:" + serviceNodeKey);
        }
        serverNodes.remove(serviceNodeKey);
        
        //
        for (Iterator<Entry<String, ServiceDefinitionGroup>> iterator = services.entrySet()
                                                                                .iterator(); iterator.hasNext();) {
            Entry<String, ServiceDefinitionGroup> entry = iterator.next();
            entry.getValue().remove(address, port);
            if (entry.getValue().isEmpty()) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Remote Service Unregistered:" + entry.getKey());
                }
                iterator.remove();
            }
        }
    }
    
    /**
     * 
     */
    public RemoteServiceDefinitionPack lookup(String serviceName,
                                              String methodName,
                                              long lastTimeMillis) {
        throw new UnsupportedOperationException();
    }
    
    /**
     * if the lastTimeMillis <= 0, always return the registered service
     * definition,otherwise,compare the lastTimeMillis to registered service
     * definition,if < the current value, it means the registered service
     * definition has changed,need return the new value to cient which will make
     * upgrade against it.
     * 
     * @param serviceName
     * @param lastTimeMillis
     * @return
     */
    public RemoteServiceDefinitionPack lookup(String serviceName,
                                              long lastTimeMillis) {
        if (StringUtils.isEmpty(serviceName)) {
            return null;
        }
        
        ServiceDefinitionGroup sdg = services.get(serviceName);
        if (sdg == null) {
            return null;
        }
        
        if (lastTimeMillis <= 0) {
            return new RemoteServiceDefinitionPack(serviceName,
                                                   true,
                                                   sdg.getLastModifiedMillis(),
                                                   sdg.toServiceDefinitionArray());
        }
        
        if (lastTimeMillis >= sdg.getLastModifiedMillis()) {
            return RemoteServiceDefinitionPack.emptyInstance();
        }
        else {
            return new RemoteServiceDefinitionPack(serviceName,
                                                   true,
                                                   sdg.getLastModifiedMillis(),
                                                   sdg.toServiceDefinitionArray());
        }
    }
    
    /**
     * 
     */
    public boolean isAvailabe(String address, int port) {
        String serviceNodeKey = address;
        if (StringUtils.isEmpty(serviceNodeKey)) {
            throw new IllegalArgumentException("The remoteAddress can't be null or empty!");
        }
        if (port != 80) {
            serviceNodeKey = new StringBuilder(address).append(':')
                                                      .append(port)
                                                      .toString();
        }
        
        ServiceNode sn = serverNodes.get(serviceNodeKey);
        if (sn == null) {
            return false;
        }
        return sn.isAvailable();
    }
    
    /**
     * 
     */
    public void register(RegistryEventListener listener) {
        RpcEventCenter.getInstance().registerListener(listener);
    }
    
    /**
     * 
     */
    public void unregister(RegistryEventListener listener) {
        RpcEventCenter.getInstance().unregisterListener(listener);
    }
    
    /**
     * @param node
     * @param listener
     */
    public void subscribe(final String address,
                          final int port,
                          final NodeEventListener listener) {
        String serviceNodeKey = address;
        if (StringUtils.isEmpty(serviceNodeKey)) {
            throw new IllegalArgumentException("The remoteAddress can't be null or empty!");
        }
        if (port != 80) {
            serviceNodeKey = new StringBuilder(address).append(':')
                                                      .append(port)
                                                      .toString();
        }
        ServiceNode sn = serverNodes.get(serviceNodeKey);
        if (sn == null) {
            return;
        }
        
        RpcEventCenter.getInstance().registerListener(new RpcEventListener() {
            
            public boolean support(RpcEventObject event) {
                if (!listener.support(event)) {
                    return false;
                }
                if (!(event instanceof NodeEventObject)) {
                    return false;
                }
                
                NodeEventObject nodeEventObject = (NodeEventObject) event;
                ServiceNode serviceNode = (ServiceNode) nodeEventObject.getSource();
                return (serviceNode.getAddress().equals(address) && serviceNode.getPort() == port);
            }
            
            public void handle(RpcEventObject event) {
                listener.handle(event);
            }
            
        });
    }
    
    /**
     * @param node
     * @param listener
     */
    public void unsubscribe(String address, int port, NodeEventListener listener) {
        throw new UnsupportedOperationException();
    }
    
    /**
     * @param eventObj
     */
    public void fireServiceEnabled(RegistryEventObject eventObject) {
        RpcEventCenter.getInstance().publishEvent(eventObject);
    }
    
    /**
     * @param eventObj
     */
    public void fireServiceDisabled(RegistryEventObject eventObject) {
        RpcEventCenter.getInstance().publishEvent(eventObject);
    }
    
    /**
     * 
     */
    public void accept(ServiceNodeVisitor visitor) {
        // TODO the unregisterAll method will remove item of serverNodes
        // the iterator logic to visit the item of serverNodes must consider it
        for (Iterator<DefaultServiceNode> iterator = serverNodes.values()
                                                                .iterator(); iterator.hasNext();) {
            visitor.visit(iterator.next());
        }
    }
    
    /**
     * 
     */
    public void shutdown() {
        if (heatBeatMonitor != null) {
            try {
                logger.debug("DefaultRegistry shuting down");
                heatBeatMonitor.stop();
                logger.debug("DefaultRegistry shuted down");
            }
            catch (Exception e) {
                e.printStackTrace();
            }
            heatBeatMonitor = null;
        }
    }
    
}
